package cn.xx.mq;

import cn.xx.dao.StockLogDOMapper;
import cn.xx.dataobject.StockLogDO;
import cn.xx.error.BusinessException;
import cn.xx.service.OrderService;
import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;

/**
 * @author xiexu
 * @create 2022-04-29 16:12
 */
@Component
public class MqProducer {

    private DefaultMQProducer producer;

    private TransactionMQProducer transactionMQProducer;

    //声明value注解，引入配置变量
    @Value("${mq.nameserver.addr}")
    private String nameAddr;

    @Value("${mq.topicname}")
    private String topicName;

    @Autowired
    private OrderService orderService;

    @Autowired
    private StockLogDOMapper stockLogDOMapper;

    @PostConstruct
    public void init() throws MQClientException {
        // 做mq producer的初始化操作
        producer = new DefaultMQProducer("producer_group");
        producer.setNamesrvAddr(nameAddr);
        producer.start();

        transactionMQProducer = new TransactionMQProducer("transaction_producer_group");
        transactionMQProducer.setNamesrvAddr(nameAddr);
        transactionMQProducer.start();
        transactionMQProducer.setTransactionListener(new TransactionListener() {

            // 发送事务型消息，消息的类型是prepare，不会被consumer立即执行
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
                // 真正要做的事，创建订单
                Integer userId = (Integer) ((Map) arg).get("userId");
                Integer itemId = (Integer) ((Map) arg).get("itemId");
                Integer promoId = (Integer) ((Map) arg).get("promoId");
                Integer amount = (Integer) ((Map) arg).get("amount");
                String stockLogId = (String) ((Map) arg).get("stockLogId");
                try {
                    // 这里进行订单的创建
                    orderService.createOrder(userId, itemId, promoId, amount, stockLogId);
                } catch (BusinessException e) {
                    e.printStackTrace();
                    // 设置对应的stockLog为回滚状态
                    StockLogDO stockLogDO = stockLogDOMapper.selectByPrimaryKey(stockLogId);
                    stockLogDO.setStatus(3);
                    stockLogDOMapper.updateByPrimaryKeySelective(stockLogDO);
                    /**
                     * 如果订单创建失败则事务回滚
                     * ROLLBACK_MESSAGE表示将之前的prepare消息撤回，相当于没发送消息
                     */
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                /**
                 * 订单创建成功，可以让消息被consumer消费
                 * COMMIT_MESSAGE表示将之前的prepare消息设置为commit，给对应的consumer去消费
                 */
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            /**
             * 消息中间件没有返回ROLLBACK_MESSAGE也没有返回COMMIT_MESSAGE的时候，有多种情况：
             * 1.对应的创建订单是成功的，已经走到（return LocalTransactionState.COMMIT_MESSAGE;）这一步，
             * 但是往中间件发送确认消息时，消息丢失了。
             * 2.对应的创建订单失败，走到（return LocalTransactionState.ROLLBACK_MESSAGE;）这一步，
             * 但是这个ROLLBACK_MESSAGE并没有发送出去。
             * 3.createOrder还没有执行完，这里就产生了回调。
             */

            /**
             * 如果上面订单的创建一直没有返回结果 或者 返回了UNKNOW，那么该方法就会被调用
             * 如果prepare状态一直没有更新，broker 会定时回查 Producer 消息状态
             * @param msg
             * @return
             */
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // 根据是否扣减库存成功，来判断要返回COMMIT，ROLLBACK还是继续UNKNOW
                String jsonString = new String(msg.getBody());
                Map<String, Object> map = JSON.parseObject(jsonString);
                Integer itemId = (Integer) map.get("itemId");
                Integer amount = (Integer) map.get("amount");
                String stockLogId = (String) map.get("stockLogId");
                StockLogDO stockLogDO = stockLogDOMapper.selectByPrimaryKey(stockLogId);
                if (stockLogDO == null) {
                    return LocalTransactionState.UNKNOW;
                }
                // 2表示下单扣减库存成功
                if (stockLogDO.getStatus().intValue() == 2) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (stockLogDO.getStatus().intValue() == 1) { //1表示初始状态
                    return LocalTransactionState.UNKNOW;
                }
                // 其他情况表示下单扣减库存失败
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });
    }

    // 事务型同步库存扣减消息
    public boolean transactionAsyncReduceStock(Integer userId, Integer itemId, Integer promoId, Integer amount, String stockLogId) {
        Map<String, Object> bodyMap = new HashMap<>();
        bodyMap.put("itemId", itemId);
        bodyMap.put("amount", amount);
        bodyMap.put("stockLogId", stockLogId);

        Map<String, Object> argsMap = new HashMap<>();
        argsMap.put("itemId", itemId);
        argsMap.put("amount", amount);
        argsMap.put("userId", userId);
        argsMap.put("promoId", promoId);
        argsMap.put("stockLogId", stockLogId);

        // 投放消息
        Message message = new Message(topicName, "increase", JSON.toJSON(bodyMap).toString().getBytes(Charset.forName("UTF-8")));
        TransactionSendResult sendResult = null;
        try {
            /**
             * sendMessageInTransaction()这个方法发送的是一个事务型消息，事务型消息有一个二阶段提交的概念。
             * 投递prepare消息，sendMessageInTransaction方法内部，就会往消息中间件投递一个对应的prepare消息，同时回调一个executeLocalTransaction方法
             * 注意这个argsMap传进去后，就会被上面的executeLocalTransaction方法的arg参数所接收
             */
            sendResult = transactionMQProducer.sendMessageInTransaction(message, argsMap);
        } catch (MQClientException e) {
            e.printStackTrace();
            return false;
        }
        if (sendResult.getLocalTransactionState() == LocalTransactionState.ROLLBACK_MESSAGE) {
            return false;
        } else if (sendResult.getLocalTransactionState() == LocalTransactionState.COMMIT_MESSAGE) {
            return true;
        } else {
            return false;
        }
    }

    // 同步库存扣减消息
    public boolean asyncReduceStock(Integer itemId, Integer amount) {
        Map<String, Object> bodyMap = new HashMap<>();
        bodyMap.put("itemId", itemId);
        bodyMap.put("amount", amount);
        // 投放消息
        Message message = new Message(topicName, "increase", JSON.toJSON(bodyMap).toString().getBytes(Charset.forName("UTF-8")));
        try {
            // 消息发送出去，consumer端就会得到消费的通知，然后去执行
            producer.send(message, 10000);
        } catch (MQClientException e) {
            e.printStackTrace();
            return false;
        } catch (RemotingException e) {
            e.printStackTrace();
            return false;
        } catch (MQBrokerException e) {
            e.printStackTrace();
            return false;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

}
